package org.yonggan.dmp.report

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 统计各省市数据量分布情况
  * 将统计的结果输出成 json 格式，并输出到磁盘目录。
  * {"ct":943,"provincename":"内蒙古自治区","cityname":"阿拉善盟"}
  * {"ct":578,"provincename":"内蒙古自治区","cityname":"未知"}
  * {"ct":262632,"provincename":"北京市","cityname":"北京市"}
  *   要求二： 将结果写到到 mysql 数据库。
  *   要求三： 用 spark 算子的方式实现上述的统计，存储到磁盘。
  */
object RptProCityCountSQL {

  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()
      .setAppName("统计各省市数据量分布情况")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 创建SqlContext
    val sqlContext = new SQLContext(sc)

    // 先读取文件
    val baseDf = sqlContext.read.parquet("D:\\project\\5")

    // 需求一 转换成为json
    import org.apache.spark.sql.functions._
    // 读取结果
    val resDF: DataFrame = baseDf.groupBy("provincename","cityname")
                .agg(count("*") as( "cnt" )).select("provincename","cityname","cnt")


    // 转发为json
    // 方式1 DataFrame 直接
    //resDF.coalesce(1).write.mode(SaveMode.Overwrite).json("d:/project/6")

    //将数据写入MySQL    val prop = new Properties()
    val prop = new Properties()
    prop.setProperty("user", "root")
    prop.setProperty("password", "1234")
    prop.setProperty("driver","com.mysql.jdbc.Driver")
    resDF.write.mode(SaveMode.Append)
      .jdbc("jdbc:mysql://localhost:3306/dmp?characterEncoding=utf8","dmp_pro_city",prop)

    sc.stop()
  }

}
